package com.avengers.core.thread;

import com.jfinal.plugin.activerecord.Page;
import org.slf4j.Logger;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.concurrent.ExecutorService;

import static com.avengers.core.bean.Numbers.INT_1;


/**
 * 多线程任务处理
 *
 * @author yu.wang
 * @since 2018-05-18 下午2:33
 */
public abstract class AbstractThreadJobHandler<T> {
    private static final Integer DEFAULT_PAGE_SIZE = 1000;
    private Integer pageSize;
    private String jobName;
    private Integer threadCount;

    public AbstractThreadJobHandler(String jobName, int threadCount) {
        this(jobName, threadCount, DEFAULT_PAGE_SIZE);
    }

    public AbstractThreadJobHandler(String jobName, int threadCount, int pageSize) {
        this.jobName = jobName;
        this.threadCount = threadCount;
        this.pageSize = pageSize;
    }

    public final void start() {
        Logger logger = logger();
        Long lastBeginNanoTime = getCurrentJobNo();
        if (null != getCurrentJobNo()) {
            logger.info("------ 多线程定时任务【{} - {}】仍未完成, 已用时：{}，终止本次任务! ------", jobName, lastBeginNanoTime,
                    BigDecimal.valueOf((System.nanoTime() - lastBeginNanoTime) / 1000000000.0).setScale(INT_1, RoundingMode.HALF_UP));
            return;
        }
        ExecutorService pool = ThreadPoolFactory.create(threadCount);
        long beginNanoTime = System.nanoTime();
        logger.info("*********** 多线程定时任务【{}】准备执行，任务号：{} ***********", jobName, beginNanoTime);
        beginJob(beginNanoTime);
        int totalExecuteCount = 0;
        int totalExcludeCount = 0;
        try {
            int pageNumber = 1;
            Page<T> page;
            do {
                page = queryPage(pageNumber++, pageSize);
                if (null != page && null != page.getList()) {
                    logger.info(">>>>>>>>> 多线程定时任务【{} - {}】处理第 {} 页 共 {}/{} 条记录", jobName, beginNanoTime, page.getPageNumber(), page.getList().size(), page.getTotalRow());
                    int executeCount = 0;
                    int excludeCount = 0;
                    for (T t : page.getList()) {
                        if (!exclude(t)) {
                            executeCount++;
                            pool.execute(() -> {
                                try {
                                    handlerEach(t);
                                } catch (Exception e) {
                                    logger.error("多线程定时任务【{} - {} ({})】执行异常", jobName, beginNanoTime, eachOneErrorRemark(t), e);
                                }
                            });
                        } else {
                            excludeCount++;
                        }
                    }
                    totalExecuteCount += executeCount;
                    totalExcludeCount += excludeCount;
                    logger.info(">>>>>>>>> 多线程定时任务【{} - {}】处理第 {} 页 共 {}/{} 条记录, 同步 {} 条，排除 {} 条", jobName, beginNanoTime, page.getPageNumber(), page.getList().size(), page.getTotalRow(), executeCount, excludeCount);
                }
            } while (null != page && !page.isLastPage() && page.getTotalPage() > 0);
        } catch (Exception e) {
            logger.error("多线程定时任务【{} - {}】执行异常", jobName, beginNanoTime, e);
        } finally {
            pool.shutdown();
            while (true) {
                if (pool.isTerminated()) {
                    endJob();
                    logger.info("*********** 多线程定时任务【{} - {}】，所有子线程执行完成，共同步{}条，排除{}条，共耗时 {} 秒! ***********",
                            jobName, beginNanoTime, totalExecuteCount, totalExcludeCount, BigDecimal.valueOf((System.nanoTime() - beginNanoTime) / 1000000000.0).setScale(INT_1, RoundingMode.HALF_UP));
                    break;
                }
            }
        }
    }

    /**
     * 分页
     *
     * @param pageNumber 页码
     * @param pageSize   数量
     * @return 分页对象
     */
    protected abstract Page<T> queryPage(int pageNumber, int pageSize);

    /**
     * jobNo
     *
     * @return long
     */
    protected abstract Long getCurrentJobNo();

    /**
     * 任务开始前执行
     *
     * @param jobNo 任务号
     */
    protected abstract void beginJob(Long jobNo);

    /**
     * 任务结束前执行
     */
    protected abstract void endJob();

    /**
     * logger
     *
     * @return logger
     */
    protected abstract Logger logger();

    /**
     * 数据处理
     *
     * @param t 数据内容
     */
    protected abstract void handlerEach(T t);

    /**
     * 是否排除
     *
     * @param t 数据内容
     * @return true|false
     */
    protected abstract boolean exclude(T t);

    /**
     * 错误处理标记
     *
     * @param t 数据内容
     * @return remark
     */
    protected abstract String eachOneErrorRemark(T t);
}
